package com.icbc.dccsh.storm.mongodb.bolt;

import java.util.Map;

import org.apache.commons.lang.Validate;

import com.icbc.dccsh.storm.mongodb.common.MongoDBClient;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.base.BaseRichBolt;

public abstract class AbstractMongoBolt extends BaseRichBolt{
	private String url;
//    private String collectionName;

    protected OutputCollector collector;
    protected MongoDBClient mongoClient;

    public AbstractMongoBolt(String url) {
       Validate.notEmpty(url, "url can not be blank or null");
//       Validate.notEmpty(collectionName, "collectionName can not be blank or null");

       this.url = url;
//       this.collectionName = collectionName;
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context,
            OutputCollector collector) {
        this.collector = collector;
//        this.mongoClient = new MongoDBClient(url, collectionName);
        this.mongoClient = new MongoDBClient(url);
    }

    @Override
    public void cleanup() {
       this.mongoClient.close();
    }
}
